package com.atguigu.day04;

import com.atguigu.utils.ClickEvent;
import com.atguigu.utils.ClickSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;

// 触发器
// 窗口的第一条数据到达以后，第一条数据的时间戳后面的整数秒，触发窗口计算
public class Example7 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<ClickEvent>forMonotonousTimestamps()
                .withTimestampAssigner(new SerializableTimestampAssigner<ClickEvent>() {
                    @Override
                    public long extractTimestamp(ClickEvent element, long recordTimestamp) {
                        return element.ts;
                    }
                }))
                .keyBy(r -> r.url)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .trigger(new MyTrigger())
                .aggregate(
                        new AggregateFunction<ClickEvent, Long, Long>() {
                            @Override
                            public Long createAccumulator() {
                                return 0L;
                            }

                            @Override
                            public Long add(ClickEvent value, Long accumulator) {
                                return accumulator + 1L;
                            }

                            @Override
                            public Long getResult(Long accumulator) {
                                return accumulator;
                            }

                            @Override
                            public Long merge(Long a, Long b) {
                                return null;
                            }
                        },
                        new ProcessWindowFunction<Long, String, String, TimeWindow>() {
                            @Override
                            public void process(String s, Context context, Iterable<Long> elements, Collector<String> out) throws Exception {
                                out.collect("url是：" + s + "的窗口" + new Timestamp(context.window().getStart()) + "~" +
                                        "" + new Timestamp(context.window().getEnd()) + "的pv次数是：" + elements.iterator().next());
                            }
                        }
                )
                .print();

        env.execute();
    }

    public static class MyTrigger extends Trigger<ClickEvent, TimeWindow> {
        // 每来一条窗口中的数据就触发一次
        @Override
        public TriggerResult onElement(ClickEvent element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
            // 当前窗口的状态变量
            // 标志位
            ValueState<Boolean> flag = ctx.getPartitionedState(new ValueStateDescriptor<Boolean>("flag", Types.BOOLEAN));
            if (flag.value() == null) {
                // 窗口的第一条数据到达
                // 1234 => 2000
                // 数据的时间戳接下来的整数秒
                long nextTs = element.ts + 1000L - element.ts % 1000L;
                // 注册的定时器是onEventTime方法
                ctx.registerEventTimeTimer(nextTs);
                // 将标志位置为true，保证只有第一条数据到达时，才会进入if语句
                flag.update(true);
            }
            return TriggerResult.CONTINUE;
        }

        // 机器时间到达time时，触发调用
        @Override
        public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            return TriggerResult.CONTINUE;
        }

        // 水位线的到达time时，触发调用
        @Override
        public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            if (time < window.getEnd()) {
                // 保证定时器时间戳小于窗口结束时间
                if (time + 1000L < window.getEnd()) {
                    // 注册的还是onEventTime
                    ctx.registerEventTimeTimer(time + 1000L);
                }
                return TriggerResult.FIRE; // 触发窗口的aggregate的执行
            }
            return TriggerResult.CONTINUE;
        }

        // 窗口关闭时，触发调用
        @Override
        public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
            ValueState<Boolean> flag = ctx.getPartitionedState(new ValueStateDescriptor<Boolean>("flag", Types.BOOLEAN));
            flag.clear();
        }
    }
}
